package drds.data_migrate.task_group.task.run;

import drds.common.Else;
import drds.data_migrate.task_group.task.data_exchange.exchanger.Produce;
import drds.data_migrate.task_group.task.data_exchange.record.Record;
import drds.data_migrate.task_group.task.data_exchange.record.column.*;
import drds.data_migrate.util.Jdbc;
import drds.data_migrate.util.Jdbcs;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;

/**
 * Created by ibm on 2018/11/3.
 */
@Slf4j
public class ReadTask
{


    @Setter
    @Getter
    private TaskConfig taskConfig;


    public ReadTask(TaskConfig taskConfig)
    {

        this.taskConfig = taskConfig;

    }

    public void init()
    {


    }

    public void startRead(Produce produce) throws SQLException
    {
        Connection connection = Jdbc.getConnection(taskConfig.getJdbcUrl(), taskConfig.getUsername(), taskConfig.getPassword());
        // set session config
        Jdbcs.sessionSet(connection, taskConfig.getSessionSqlList());

        int columnNumber = 0;
        ResultSet resultSet = null;
        try
        {
            resultSet = Jdbc.executeQuery(connection, taskConfig.getQuerySql());
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            columnNumber = resultSetMetaData.getColumnCount();
            while (resultSet.next())
            {
                this.transportOneRecord(produce, resultSet, resultSetMetaData, columnNumber);
            }
        } catch (SQLException e)
        {
            throw e;
        } finally
        {
            Jdbc.close(null, connection);
        }
    }

    public void post()
    {
        // do nothing
    }

    public void destroy()
    {
        // do nothing
    }

    protected Record transportOneRecord(Produce produce, ResultSet resultSet, ResultSetMetaData resultSetMetaData, int columnNumber)
    {
        Record record = buildRecord(produce, resultSet, resultSetMetaData, columnNumber);
        produce.addToRecordList(record);
        return record;
    }

    /**
     * <pre>
     * int sqlType = getColumnType(columnIndex);
     *
     * switch (sqlType) {
     *
     * case Types.NUMERIC:
     * case Types.DECIMAL:
     * className = java.math.BigDecimal.class.getName();
     * break;
     *
     * case Types.BIT:
     * className = java.lang.Boolean.class.getName();
     * break;
     *
     * case Types.TINYINT:
     * className = java.lang.Byte.class.getName();
     * break;
     *
     * case Types.SMALLINT:
     * className = java.lang.Short.class.getName();
     * break;
     *
     * case Types.INTEGER:
     * className = java.lang.Integer.class.getName();
     * break;
     *
     * case Types.BIGINT:
     * className = java.lang.Long.class.getName();
     * break;
     *
     * case Types.REAL:
     * className = java.lang.Float.class.getName();
     * break;
     *
     * case Types.FLOAT:
     * case Types.DOUBLE:
     * className = java.lang.Double.class.getName();
     * break;
     *
     * case Types.BINARY:
     * case Types.VARBINARY:
     * case Types.LONGVARBINARY:
     * className = "byte[]";
     * break;
     *
     * case Types.DATE:
     * className = java.sql.Date.class.getName();
     * break;
     *
     * case Types.TIME:
     * className = java.sql.Time.class.getName();
     * break;
     *
     * case Types.TIMESTAMP:
     * className = java.sql.Timestamp.class.getName();
     * break;
     *
     * case Types.BLOB:
     * className = java.sql.Blob.class.getName();
     * break;
     *
     * case Types.CLOB:
     * className = java.sql.Clob.class.getName();
     * break;
     * }
     * return className;
     * </pre>
     */


    protected Record buildRecord(Produce produce, ResultSet resultSet, ResultSetMetaData resultSetMetaData, int columnNumber)
    {
        Record record = produce.createRecord();

        try
        {
            for (int i = 1; i <= columnNumber; i++)
            {


                switch (resultSetMetaData.getColumnClassName(i))//采用类名  比类型值更好
                {
                    case "java.lang.Boolean":
                        record.addColumn(new BooleanColumn(resultSet.getBoolean(i)));
                        break;
                    //
                    case "java.lang.Byte":
                        record.addColumn(new ByteColumn(resultSet.getByte(i)));
                        break;
                    case "java.lang.Short":
                        record.addColumn(new ShortColumn(resultSet.getShort(i)));
                        break;
                    case "java.lang.Integer":
                        record.addColumn(new IntegerColumn(resultSet.getInt(i)));
                        break;
                    case "java.lang.Long":
                        record.addColumn(new LongColumn(resultSet.getLong(i)));
                        break;
                    case "java.lang.Float":
                        record.addColumn(new FloatColumn(resultSet.getFloat(i)));
                        break;
                    case "java.lang.Double":
                        record.addColumn(new DoubleColumn(resultSet.getDouble(i)));
                        break;

                    case "java.math.BigDecimal":
                        record.addColumn(new BigDecimalColumn(resultSet.getBigDecimal(i)));
                        break;
                    //
                    case "java.lang.String":
                        record.addColumn(new StringColumn(resultSet.getString(i)));
                        break;
                    //
                    case "java.sql.Date":
                        record.addColumn(new DateColumn(resultSet.getDate(i)));
                        break;
                    case "java.sql.Time":
                        record.addColumn(new TimeColumn(resultSet.getTime(i)));
                        break;
                    case "java.sql.Timestamp":
                        record.addColumn(new TimestampColumn(resultSet.getTimestamp(i)));
                        break;
                    //
                    case "byte[]":
                    case "java.sql.Blob":
                    case "java.sql.Clob":
                    default:
                        throw new Else();
                }
            }
        } catch (Exception e)
        {

            //TODO 这里识别为脏数据靠谱吗？
            //abstractTaskPluginCollector.collectDirtyRecord(record, e);
            throw new RuntimeException(e);
        }
        return record;
    }
}
